package com.atguigu.datastream.test.day04;

import com.atguigu.datastream.bean.Event;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * ClassName: Flink_05_Sink_mysql_Customer
 * Package: com.atguigu.day04
 * Description:
 *
 * @Author ChenJun
 * @Create 2023/4/10 19:12
 * @Version 1.0
 */
public class Flink_05_Sink_mysql_Customer {
    public static void main(String[] args) throws Exception {

        //1. 创建流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.获取数据
        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);


        SingleOutputStreamOperator<Event> map = streamSource.map(new MapFunction<String, Event>() {
            @Override
            public Event map(String value) throws Exception {
                String[] split = value.split(",");
                return new Event(split[0], split[1], Long.parseLong(split[2]));
            }
        });

        map.addSink(new MySink());
        env.execute();

    }

    public static class MySink extends RichSinkFunction<Event> {

        private Connection connection;
        private PreparedStatement pstt;

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("创建链接");
            //1. 创建链接
            connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test", "root", "000000");

            //2. 创建语句执行者
            pstt = connection.prepareStatement("insert into clicks values(?,?)");
        }

        @Override
        public void invoke(Event value, Context context) throws Exception {

            pstt.setString(1, value.user);
            pstt.setString(2, value.url);
            pstt.execute();

        }


        @Override
        public void close() throws Exception {
            System.out.println("关闭链接");
            pstt.close();
            connection.close();
        }
    }
}
